package com.abyss.source;


import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

import java.util.UUID;
import java.util.stream.Stream;

/**
 * 实现一个带有rich功能的 可并行的source
 */
public class CustomRichParallelSourceDemo {
    public static void main(String[] args) throws Exception {
        // 1. env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 2. addSource
        DataStreamSource<String> source = env.addSource(new MyRichSourceWithParallel()).setParallelism(6);

        source.print();
        env.execute();
    }

    public static class MyRichSourceWithParallel extends RichParallelSourceFunction<String> {
        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            System.out.println(getRuntimeContext().getIndexOfThisSubtask() + ";im open");
        }

        @Override
        public void close() throws Exception {
            super.close();
            System.out.println(getRuntimeContext().getIndexOfThisSubtask() + ";im closed");
        }

        @Override
        public void run(SourceContext<String> ctx) throws Exception {
            ctx.collect(UUID.randomUUID().toString());
        }

        @Override
        public void cancel() {

        }
    }
}